package com.gitee.sparta.core.process.workers;

import cn.hutool.core.bean.BeanUtil;
import com.fasterxml.jackson.core.JsonProcessingException;
import com.gitee.sparta.common.exception.SpartaException;
import com.gitee.sparta.common.helpers.ExecuteInvokeHelper;
import com.gitee.sparta.common.helpers.NetworkDelayHelper;
import com.gitee.sparta.common.helpers.RedisKeyHelper;
import com.gitee.sparta.core.bean.SpartaBeanMapping;
import com.gitee.sparta.core.bean.SpartaChannel;
import com.gitee.sparta.core.bean.metadata.Metadata;
import com.gitee.sparta.core.process.SpartaThreadPoolFactory;
import com.gitee.sparta.core.process.feature.SpartaAcknowledgeFeature;
import com.gitee.sparta.core.process.feature.SpartaDistributedRetryFeature;
import com.gitee.sparta.core.process.operate.MetadataTabOperate;
import com.gitee.sparta.core.process.operate.ReadableTabOperate;
import com.gitee.sparta.core.property.SpartaCoreProperties;
import lombok.AllArgsConstructor;
import lombok.extern.slf4j.Slf4j;

import java.lang.reflect.InvocationTargetException;
import java.time.Duration;
import java.util.Iterator;
import java.util.Map;
import java.util.Objects;
import java.util.Set;
import java.util.concurrent.FutureTask;
import java.util.concurrent.TimeUnit;

/**
 * 监听器工作者
 * 主要工作内容，分发任务，并执行任务
 *
 * @author zyred
 * @since v 0.0.1
 */
@Slf4j
@AllArgsConstructor
public class SpartaListenWorker<V> {

    private final SpartaCoreProperties spartaProperties;
    private final ReadableTabOperate<V> readableTabOperate;
    private final MetadataTabOperate<V> metadataTabOperate;
    private final SpartaThreadPoolFactory spartaThreadPoolFactory;
    private final SpartaDistributedRetryFeature<V> retryFeature;
    private final SpartaAcknowledgeFeature<V> spartaAcknowledgeFeature;

    /** 启动开关 **/
    private static volatile boolean enable = true;


    /**
     * 上游业务处理完毕后，将业务分发到此处进行调用用户的业务代码
     *
     * @param listenTopic   待消费的 TOPIC，也就是 @SpartaListener 的 topics 中定义的内容
     * @param mapping       @SpartaListener 所在类与方法之间的映射关系，在使用反射的时候会使用到类对象
     */
    @SuppressWarnings("all")
    public void doWorker(String listenTopic, SpartaBeanMapping mapping) {
        final String readableTabKey = RedisKeyHelper.readableKey(this.spartaProperties.getSystemName(),
                this.spartaProperties.getIsolation(), listenTopic);
        Map<String, Metadata<V>> metadataTab = this.metadataTabOperate.readMetadataMap (listenTopic);

        Set<Map.Entry<String, Metadata<V>>> entries = metadataTab.entrySet();
        Iterator<Map.Entry<String, Metadata<V>>> ite = entries.iterator();


        while (ite.hasNext()) {
            if (!enable) {
                return;
            }
            // topic:metaId
            Map.Entry<String, Metadata<V>> metadataEntry = ite.next();
            Metadata<V> metadata = metadataEntry.getValue();
            String fullTopic = metadataEntry.getKey();

            if (Objects.isNull(metadata)) {
                this.readableTabOperate.delReadableTab(fullTopic, readableTabKey);
                continue;
            }

            // 计算实际与理论的偏差
            if(!metadata.isSegment()) {
                int deviation = (int) (System.currentTimeMillis() - metadata.getExpire());
                NetworkDelayHelper.write(deviation);
            }

            V body = metadata.getBody();
            V deep = (V) BeanUtil.copyProperties(body, body.getClass());

            metadata.setExpire(metadata.isSegment() ? System.currentTimeMillis() : metadata.getExpire());

            SpartaChannel spartaChannel = new SpartaChannel(SpartaAcknowledgeFeature.getAckTag(), metadata.getMetaId(), metadata.getExpire());
            spartaChannel.setRetryIntervalTime(metadata.getRetryIntervalTime());
            spartaChannel.setRenewIntervalTime(metadata.getRenewIntervalTime());
            try {
                this.doCallback (metadata, spartaChannel, mapping);
            } catch (Exception ex) {
                // 执行重试
                metadata.setBody(deep);
                this.retryFeature.retry(fullTopic, listenTopic, metadata, spartaChannel);
                continue;
            }

            ite.remove();
            metadata.setBody(deep);
            this.spartaAcknowledgeFeature.doProcessAcknowledge(fullTopic, listenTopic, metadata, spartaChannel);
        }
    }

    /**
     * 该方法是主动调用用户的 @SpartaListener 所标注的方法
     *
     * @param metadata          元数据信息，该类是插入的时候就进行包装的，里面包含了用户的 body
     * @param spartaChannel     该对象是提供了 ack 等信息
     * @param mapping           标注了 @SpartaListener 的对象与方法的映射
     * @throws Exception        调用用户的业务异常，作为重试的依据
     */
    @SuppressWarnings("all")
    private void doCallback(Metadata<V> metadata, SpartaChannel spartaChannel, SpartaBeanMapping mapping) throws Exception {
        // 解析参数
        Object[] params = ExecuteInvokeHelper.parseTargetMethod(metadata, spartaChannel, mapping);

        ThreadLocal<Integer> featureAckTag = SpartaAcknowledgeFeature.getAckTag();
        final Duration waitTime = this.spartaProperties.getConsumeWaitTime();

        // 关闭了业务的超时等待时间
        if (Duration.ZERO.equals(waitTime)) {
            try {
                mapping.getCallbackMethod().invoke(mapping.getSpringBean(), params);
                return;
            } catch (Exception ex) {
                log.error("[Sparta] Callback business fail, exception: {}", ex.getMessage(), ex);
                throw ex;
            }
        }

        // 开启了业务超时等待时间
        try {
            FutureTask<Integer> callbackFuture = new FutureTask<>(() -> {
                mapping.getCallbackMethod().invoke(mapping.getSpringBean(), params);
                // 标识上推
                return SpartaAcknowledgeFeature.removeAck();
            });
            this.spartaThreadPoolFactory.getTimeoutThreadPool().execute(callbackFuture);
            // 同步结果，超时异常
            featureAckTag.set(callbackFuture.get(waitTime.getSeconds(), TimeUnit.SECONDS));
        } catch (Exception ex) {
            log.error("[Sparta] Timeout call business fail, exception: {}", ex.getMessage(), ex);
            throw ex;
        }
    }


    /**
     * 1. 修改循环标识，停止对读取的内容进行处理
     * 2. 管理监听处理线程
     * 3. 关闭超时处理的线程
     */
    public void shutdown () {
        enable = false;
        this.spartaThreadPoolFactory.getListenersThreadPool().shutdown();
        this.spartaThreadPoolFactory.getTimeoutThreadPool().shutdown();
    }

    /**
     * 执行广播回调
     *
     * @param mapping       广播对应的 spring bean 与 method
     * @param metadata      元数据信息
     */
    public void broadcast(SpartaBeanMapping mapping, Metadata<V> metadata) {
        try {
            Object[] params = ExecuteInvokeHelper.parseTargetMethod(metadata, null, mapping);
            mapping.getCallbackMethod().invoke(mapping.getSpringBean(), params);
        } catch (InvocationTargetException | IllegalAccessException | JsonProcessingException ex) {
            log.error("[Sparta] Call business fail, exception: {}", ex.getMessage(), ex);
            throw new SpartaException(ex);
        }
    }
}
